package net.neptech.websocket;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import net.neptech.entity.vo.dto.KSignalMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.kafka.support.SendResult;
import java.io.IOException;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

/**
 * WebSocket 信令服务器 (带 Session 级别锁以解决并发)
 */
@ServerEndpoint(value = "/websocket/{conferenceId}/{userId}")
@Component
public class SignalingWebSocket {
    // 先不初始化，先声明
    private static KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public void setKafkaTemplate(KafkaTemplate<String, String> kafkaTemplate) {
        SignalingWebSocket.kafkaTemplate = kafkaTemplate;
    }

    // 存储所有活跃的会话: Key = "conferenceId:userId", Value = Session
    private static final Map<String, Session> sessions = new ConcurrentHashMap<>();

    // 管理每个会议房间的用户列表: Key = conferenceId, Value = Set of userIds
    private static final Map<String, Set<String>> conferenceRooms = new ConcurrentHashMap<>();

    // 为每个 Session 分配一个锁，用于同步发送操作，防止 IllegalStateException
    private static final Map<Session, ReentrantLock> sessionLocks = new ConcurrentHashMap<>();

    /**
     * 用户连接建立
     */
    @OnOpen
    public void onOpen(Session session,
                       @PathParam("conferenceId") String conferenceId,
                       @PathParam("userId") String userId) {
        System.out.println("🟢 用户连接成功: userId=" + userId + ", conferenceId=" + conferenceId);

        String key = conferenceId + ":" + userId;
        sessions.put(key, session);
        // 为新建立的 Session 创建并存储一个锁
        sessionLocks.put(session, new ReentrantLock());
        conferenceRooms.computeIfAbsent(conferenceId, k -> new HashSet<>()).add(userId);
        System.out.println("🎉 用户加入: " + key + " | 房间人数: " + conferenceRooms.get(conferenceId).size());
        try {
            // 发送欢迎消息给自己
            session.getBasicRemote().sendText("{\"type\":\"welcome\",\"msg\":\"欢迎加入会议: " + conferenceId + "\"}");
            //广播：通知房间内其他所有人“我来了”
            JSONObject joinMsg = new JSONObject();
            joinMsg.put("type", "user-joined");
            joinMsg.put("userId", userId);
            joinMsg.put("timestamp", System.currentTimeMillis());
            broadcast(conferenceId, userId, joinMsg.toJSONString());

            // 主动告诉新用户：房间里已经有哪些人
            Set<String> existingUsers = conferenceRooms.get(conferenceId);
            if (existingUsers != null) {
                existingUsers.stream()
                        .filter(id -> !id.equals(userId)) // 排除自己
                        .forEach(otherUserId -> {
                            try {
                                JSONObject existingMsg = new JSONObject();
                                existingMsg.put("type", "existing-user");
                                existingMsg.put("userId", otherUserId);
                                // 发送消息时也需要加锁
                                sendMessageSafely(session, existingMsg.toJSONString());
                            } catch (Exception e) { // 捕获所有异常，防止影响其他用户
                                System.err.println("发送 existing-user 消息失败给 " + userId + ": " + e.getMessage());
                            }
                        });
            }
        } catch (Exception e) { // 捕获所有异常
            System.err.println("发送欢迎消息失败: " + e.getMessage());
        }
    }

    /**
     * 向房间内所有其他用户广播消息 (同样需要对每个接收者加锁)
     */
    public static void broadcast(String conferenceId, String senderId, String message) {
        Set<String> members = conferenceRooms.get(conferenceId);
        if (members == null) return;

        members.forEach(userId -> {
            if (userId.equals(senderId)) return; // 不发给自己
            String key = conferenceId + ":" + userId;
            Session session = sessions.get(key);
            if (session != null && session.isOpen()) {
                try {
                    sendMessageSafely(session, message);
                } catch (Exception e) {
                    System.err.println("广播消息失败给 " + key + ": " + e.getMessage());
                }
            }
        });
    }

    /**
     * 接收来自前端的webSocket消息，转发进kafka
     */
    @OnMessage
    public void onMessage(String message, Session session,
                          @PathParam("conferenceId") String conferenceId,
                          @PathParam("userId") String userId) {
        try {
            JSONObject json = JSON.parseObject(message);
            String type = json.getString("type");
            String toUserId = json.getString("to");
            Object payload = json.get("data");


            KSignalMessage kMsg = new KSignalMessage();
            kMsg.setConferenceId(conferenceId);
            kMsg.setFromUserId(userId);
            kMsg.setToUserId(toUserId);
            kMsg.setType(type);
            kMsg.setData(payload);

            kafkaTemplate.send("lxy-signaling-top", JSON.toJSONString(kMsg));

            // 2) 再“本机直转”：确保同实例内的对端立即收到
            JSONObject forward = new JSONObject();
            forward.put("type", type);
            forward.put("from", userId);
            forward.put("data", payload);
            if ("file-shared".equals(type) || "all".equalsIgnoreCase(toUserId)) {
                broadcast(conferenceId, userId, forward.toJSONString());
            } else {
                String toKey = conferenceId + ":" + toUserId;
                Session target = sessions.get(toKey);
                if (target != null && target.isOpen()) {
                    sendMessageSafely(target, forward.toJSONString());
                }
            }
        } catch (Exception e) {
            System.err.println("处理消息失败: " + message);
            e.printStackTrace();
            try {
                sendMessageSafely(session, "{\"type\":\"error\",\"msg\":\"服务器处理消息失败\"}");
            } catch (Exception ioException) {
                System.err.println("发送错误消息失败: " + ioException.getMessage());
            }
        }
    }

    /**
     * 供 Kafka Consumer 调用的静态方法，用于实际发送 WebSocket 消息
     */
    public static void forwardMessageFromKafka(KSignalMessage kMsg) {
        System.out.println("sessions 当前有: " + sessions.keySet());
        String conferenceId = kMsg.getConferenceId();
        String fromUserId = kMsg.getFromUserId();
        String type = kMsg.getType();
        Object data = kMsg.getData();
        JSONObject forwardMessage = new JSONObject();
        forwardMessage.put("type", type);
        forwardMessage.put("from", fromUserId);
        forwardMessage.put("data", data);
        String forwardJson = forwardMessage.toJSONString();
        // 对 file-shared 做广播
        if ("file-shared".equals(type)) {
            Set<String> members = conferenceRooms.get(conferenceId);
            if (members != null) {
                for (String userId : members) {
                    String toKey = conferenceId + ":" + userId;
                    Session session = sessions.get(toKey);
                    System.out.println("尝试转发 toKey=" + toKey);
                    if (session != null && session.isOpen()) {
                        try {
                            sendMessageSafely(session, forwardJson);
                            System.out.println("📢 (广播) 发送消息给 " + userId + ": " + forwardJson);
                        } catch (Exception e) {
                            System.err.println("广播失败给 " + toKey + ": " + e.getMessage());
                        }
                    }
                }
            }
        } else {
            String toUserId = kMsg.getToUserId();
            String toKey = conferenceId + ":" + toUserId;
            Session targetSession = sessions.get(toKey);
            if (targetSession != null && targetSession.isOpen()) {
                try {
                    sendMessageSafely(targetSession, forwardJson);
                    System.out.println("📤 (P2P) 发送给 " + toUserId + ": " + forwardJson);
                } catch (Exception e) {
                    System.err.println("P2P 发送失败: " + e.getMessage());
                }
            }
        }
    }

    /**
     * 安全地发送消息到指定的 Session
     * 通过获取与 Session 关联的锁来保证发送操作的原子性
     */
    private static void sendMessageSafely(Session session, String message) throws IOException {
        // 获取该 Session 的锁
        ReentrantLock lock = sessionLocks.get(session);
        if (lock == null) {
            throw new IOException("Session lock not found, session might be closed or removed.");
        }
        // 尝试获取锁，如果获取不到会阻塞等待
        lock.lock();
        try {
            // 在持有锁的情况下执行发送操作
            if (session.isOpen()) { // 再次检查 Session 是否仍然打开
                session.getBasicRemote().sendText(message);
            } else {
                throw new IOException("Session is closed.");
            }
        } finally {
            // 确保锁最终会被释放
            lock.unlock();
        }
    }

    public static Session getSession(String conferenceId, String userId) {
        String key = conferenceId + ":" + userId;
        return sessions.get(key);
    }
    /**
     * 用户离开
     */
    @OnClose
    public void onClose(Session session,
                        @PathParam("conferenceId") String conferenceId,
                        @PathParam("userId") String userId) {
        String key = conferenceId + ":" + userId;
        sessions.remove(key); // 移除会话

        // 移除 Session 对应的锁
        sessionLocks.remove(session);

        Set<String> members = conferenceRooms.get(conferenceId);
        if (members != null) {
            members.remove(userId);
            if (members.isEmpty()) {
                conferenceRooms.remove(conferenceId); // 房间空了就清理
            } else {
                JSONObject leaveMsg = new JSONObject();
                leaveMsg.put("type", "user-left");
                leaveMsg.put("userId", userId);
                broadcast(conferenceId, userId, leaveMsg.toJSONString());
            }
        }
        System.out.println("用户离开: " + key);
    }
    @OnError
    public void onError(Session session, Throwable error) {
        System.err.println(" WebSocket 连接错误: " + error.getMessage());
        error.printStackTrace();
        // 可以考虑清理资源
        sessionLocks.remove(session);
    }
}